#pragma once
#include "../common/net.h"
#include "../common/message.h"
#include <unordered_set>
namespace rpc
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager()
            {
            }
            void onTopicquest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                TopicOptype topic_type = msg->optype();
                bool ret = true;
                switch (topic_type)
                {
                case TopicOptype::TOPIC_CREATE:
                    topicCreate(conn, msg);
                    break;
                // 主题的删除
                case TopicOptype::TOPIC_REMOVE:
                    topicRemove(conn, msg);
                    break;
                // 主题的订阅
                case TopicOptype::TOPIC_SUBSCRIBE:
                    ret = topicSubscribe(conn, msg);
                    break;
                // 主题的取消订阅
                case TopicOptype::TOPIC_CANCEL:
                    topicCancel(conn, msg);
                    break;
                // 主题消息的发布
                case TopicOptype::TOPIC_PUBLISH:
                    ret = topicPublish(conn, msg);
                    break;
                default:
                    return errorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE);
                }
                if (!ret)
                {
                    return errorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
                }
                return topicResponse(conn, msg);
            }
            void onShutdown(const BaseConnection::ptr &conn)
            {
                // 消息发布者断开连接，不需要任何操作；  消息订阅者断开连接需要删除管理数据
                // 1. 判断断开连接的是否是订阅者，不是的话则直接返回
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _sub.find(conn);
                    if (it == _sub.end())
                    {
                        return; // 断开的连接，不是一个订阅者的连接
                    }
                    subscriber = it->second;
                    // 2. 获取到订阅者退出，受影响的主题对象
                    for (auto &topic_name : subscriber->_topic)
                    {
                        auto topic_it = _topic.find(topic_name);
                        if (topic_it == _topic.end())
                            continue;
                        topics.push_back(topic_it->second);
                    }
                    // 4. 从订阅者映射信息中，删除订阅者
                    _sub.erase(it);
                }
                // 3. 从受影响的主题对象中，移除订阅者
                for (auto &topic : topics)
                {
                    topic->eraseSubscribe(subscriber);
                }
            }

        private:
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, RCode rcode)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setrId(msg->rid());
                msg_rsp->setMtype(MType::RSP_TOPIC);
                msg_rsp->SetRCode(rcode);
                conn->send(msg_rsp);
            }
            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setrId(msg->rid());
                msg_rsp->setMtype(MType::RSP_TOPIC);
                msg_rsp->SetRCode(RCode::RCODE_OK);
                conn->send(msg_rsp);
            }
            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                std::string topic_name = msg->topic();
                auto topic = std::make_shared<Topic>(topic_name);
                _topic.insert(std::make_pair(topic_name, topic));
            }
            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                std::string topic_name = msg->topic();

                std::unique_lock<std::mutex> locK(_mutex);
                std::unordered_set<Subscriber::ptr> sub;
                {
                    auto it = _topic.find(topic_name);
                    if (it == _topic.end())
                    {
                        return;
                    }
                    sub = it->second->subscribe;
                    _topic.erase(it);
                }
                for (auto &s : sub)
                {
                    s->erasetopic(topic_name);
                }
            }
            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Subscriber::ptr sub;
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topic.find(msg->topic());
                    if (topic_it == _topic.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;
                    auto sub_it = _sub.find(conn);
                    if (sub_it == _sub.end())
                    {
                        sub = std::make_shared<Subscriber>(conn);
                        _sub.insert(std::make_pair(conn, sub));
                    }
                    else
                    {
                        sub = sub_it->second;
                    }
                }
                topic->appendSubscribe(sub);
                sub->appendtopic(msg->topic());
                return true;
            }
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Subscriber::ptr sub;
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto sub_it = _sub.find(conn);
                    if (sub_it != _sub.end())
                    {
                        sub = sub_it->second;
                    }

                    auto topic_it = _topic.find(msg->topic());
                    if (topic_it != _topic.end())
                    {
                        topic = topic_it->second;
                    }
                }
                if (sub)
                {
                    sub->erasetopic(msg->topic());
                }
                if (sub && topic)
                {
                    topic->eraseSubscribe(sub);
                }
            }
            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topic.find(msg->topic());
                    if (topic_it == _topic.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;
                }
                topic->pushMessage(msg);
                return true;
            }

        private:
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr conn;
                std::unordered_set<std::string> _topic;
                Subscriber(const BaseConnection::ptr &c)
                    : conn(c)
                {
                }
                void appendtopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> locK(_mutex);
                    _topic.insert(topic_name);
                }
                void erasetopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topic.erase(topic_name);
                }
            };
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                std::string topic_name;
                Topic(const std::string &name)
                    : topic_name(name)
                {
                }
                std::unordered_set<Subscriber::ptr> subscribe;

                void appendSubscribe(const Subscriber::ptr &sub)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribe.insert(sub);
                }
                void eraseSubscribe(const Subscriber::ptr &sub)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribe.erase(sub);
                }
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for (auto &sub : subscribe)
                    {
                        sub->conn->send(msg);
                    }
                }
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topic;
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _sub;
        };
    }
}